package com.linkstec.Callback;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class CallbackBlockingQueue implements Runnable {

	// 阻塞队列
	private static BlockingQueue<Map<String, Object>> BlockingresultSets;
	// 线程池大小设为5后期可灵活改动
	private static final int THREAD_POOL_SIZE = 1;
	// GtjaCdtEventFlowPushToKafkaQueue线程的执行器
	static ExecutorService callBackQueueEc = null;
	private static final Logger logger = LogManager.getLogger();
	
	/**
	 * 往阻塞队列里放数据
	 * 
	 * @param listData
	 */
	public static void summitTask(Map<String, Object> mapData, Callback callBack) {
		try {
			mapData.put("callBack", callBack);
			// 阻塞队列放数据
			BlockingresultSets.put(mapData);
		} catch (InterruptedException e) {
			logger.error("kafka发送类错误", e);
		}
	}

	private static void init() {
		if (null == BlockingresultSets) {
			// 长度为100的FIFO阻塞型队列
			BlockingresultSets = new LinkedBlockingDeque<Map<String, Object>>(
					10000);
		}
		if (null == callBackQueueEc) {
			callBackQueueEc = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
		}
		int i = 0;
		while (i < THREAD_POOL_SIZE) {
			callBackQueueEc.execute(new CallbackBlockingQueue());
			i++;
		}
		
	}

	@Override
	public void run() {
//		Thread.currentThread().setName("异步调用线程");
		while (true){
			Map<String, Object> activeCustMap;
			try {
//				activeCustMap = BlockingresultSets.poll();
				activeCustMap = BlockingresultSets.take();
				Callback callBack = (Callback) activeCustMap.get("callBack");
				callBack.onCompletion();
				
			} catch (Exception ie) {
				logger.error("错误", ie);
			}
			
		}
	}

	// 初始化代码块
	static {
		init();
	}
	public static void main(String[] args) {
		Map<String, Object> mapData = new HashMap<String, Object>();
		CallbackBlockingQueue.summitTask(mapData, new Callback() {
			
			@Override
			public void onCompletion() {
				logger.info("调用结束");
				
			}
		});
		logger.info("发起调用");
	}
}